package cloudy.topo;

import cloudy.bolt.AreaAmtBolt;
import cloudy.bolt.AreaFilterBolt;
import cloudy.bolt.AreaResBolt;
import cloudy.spout.OrderBaseSpout;
import kafka.examples.KafkaProperties;
import org.apache.storm.Config;
import org.apache.storm.LocalCluster;
import org.apache.storm.StormSubmitter;
import org.apache.storm.generated.AlreadyAliveException;
import org.apache.storm.generated.AuthorizationException;
import org.apache.storm.generated.InvalidTopologyException;
import org.apache.storm.topology.TopologyBuilder;
import org.apache.storm.tuple.Fields;

/**
 * @Package cloudy.topo
 * @Description:
 * @Author elwyn
 * @Date 2017/9/24 20:52
 * @Email elonyong@163.com
 */
public class AreaAmtTopo {
    public static void main(String[] args) throws Exception {
        TopologyBuilder topologyBuilder = new TopologyBuilder();
        topologyBuilder.setSpout("spout", new OrderBaseSpout(KafkaProperties.ORDER_TOPIC), 5);
        topologyBuilder.setBolt("filter", new AreaFilterBolt(), 5).shuffleGrouping("spout");
        topologyBuilder.setBolt("areabolt", new AreaAmtBolt(), 2).fieldsGrouping("filter",new Fields("area_id"));
        topologyBuilder.setBolt("resBolt",new AreaResBolt(),1).shuffleGrouping("areabolt");

        Config config = new Config();
        if (args.length > 0) {
            try {
                StormSubmitter.submitTopology(args[0], config, topologyBuilder.createTopology());
            } catch (AlreadyAliveException | InvalidTopologyException | AuthorizationException e) {
                e.printStackTrace();
            }
        } else {
            LocalCluster localCluster = new LocalCluster();
            localCluster.submitTopology("mytopology", config, topologyBuilder.createTopology());
        }
    }
}
